package com.chenjj.bigdata.flink.watermark

import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.watermark.Watermark

/**
  * 根据特殊条件生成Punctuated(非周期性的)水位线
  */
class PunctuatedAssigner extends  AssignerWithPunctuatedWatermarks[(String,Long,Int)]{

  /**
    * 定义事件时间生成策略
    * @param element
    * @param previousElementTimestamp
    * @return
    */
  override def extractTimestamp(element: (String, Long, Int), previousElementTimestamp: Long): Long = {
    element._2
  }

  /**
    * 定义水位线生成策略
    * @param lastElement
    * @param extractedTimestamp
    * @return
    */
  override def checkAndGetNextWatermark(lastElement: (String, Long, Int), extractedTimestamp: Long): Watermark = {
    if (lastElement._3 == 0){
      new Watermark(extractedTimestamp)
    }else{
      null
    }
  }
}
